package com.cl.spark.node;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.cl.spark.base.BaseSparkNode;
import com.cl.spark.dto.SparkParam;
import com.cl.spark.dto.SparkResult;
import com.cl.spark.enums.SparkNodeEnum;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.springframework.stereotype.Component;
import scala.annotation.meta.param;

import java.util.List;
import java.util.UUID;

@Component
public class SortNode extends BaseSparkNode {

    @Override
    public SparkResult process(SparkParam sparkParam) {
        Dataset<Row> dataset = sparkParam.getDatasetList().get(0);
        JSONObject expression = sparkParam.getNodeExpression();

        JSONArray fields = expression.getJSONArray("fields");

        for (int i = fields.size() - 1; i >= 0; i--) {
            JSONObject field = fields.getJSONObject(i);
            String fieldName = field.getString("fieldName");
            String sortOrder = field.getString("sortOrder");
            if ("倒序".equals(sortOrder)) {
                dataset = dataset.sort(dataset.col(fieldName).desc());
            } else {
                dataset = dataset.sort(dataset.col(fieldName).asc());
            }
        }
        return SparkResult.success(dataset);
    }


    @Override
    public SparkNodeEnum getType() {
        return SparkNodeEnum.SORT;
    }
}
